package com.orion.config;

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.common.TopicPartition;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.*;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;

/**
 * @author Administrator
 * @date 2021/12/6
 */
@Configuration
@EnableKafka
public class KafkaContainerConfig {

    @Value("${spring.kafka.listener.poll-timeout}")
    private Integer pollTimeout;

    @Value("${spring.kafka.listener.concurrency}")
    private Integer concurrency;

    @Value("${spring.kafka.listener.ack-mode}")
    private ContainerProperties.AckMode ackMode;

    @Value("${spring.kafka.listener.type}")
    private KafkaProperties.Listener.Type type;

    @Resource
    private ConsumerFactory<String, Object> consumerFactory;

    @Resource
    private KafkaTemplate kafkaTemplate;

    private final static String DLT = ".DLT";

    @Bean
    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String>
                factory = new ConcurrentKafkaListenerContainerFactory<>();
        // 设置消费者工厂
        factory.setConsumerFactory(consumerFactory);
        // 消费者组中线程数量
        factory.setConcurrency(concurrency);

        ContainerProperties containerProperties = factory.getContainerProperties();
        // 拉取超时时间
        containerProperties.setPollTimeout(pollTimeout);
        containerProperties.setAckMode(ackMode);
        if (type.equals(KafkaProperties.Listener.Type.BATCH)) {
            factory.setBatchListener(true);
            //不自定义异常处理也是无限循环
            factory.setBatchErrorHandler(batchErrorHandler());
        } else {
            factory.setBatchListener(false);
            factory.setErrorHandler(errorHandler());
        }

        return factory;
    }

    @Bean
    @ConditionalOnProperty(name = "spring.kafka.listener.type", havingValue = "batch")
    public BatchErrorHandler batchErrorHandler() {
        System.out.println("init batchErrorHandler");
        return new BatchErrorHandler() {
            //只到这方法
            @Override
            public void handle(Exception thrownException, ConsumerRecords<?, ?> data) {
               /* System.out.println(System.currentTimeMillis() + "--first here");*/
                System.out.println(thrownException.getMessage() + " --- " + data);
               //自定义转发到某个队列，再处理,那个队列单独定义异常处理器
                //https://www.jb51.net/article/216521.htm
                data.forEach(item -> {
                    ListenableFuture future = kafkaTemplate.send(item.topic() + DLT, item.key(), item.value());
                    future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
                        @Override
                        public void onFailure(Throwable ex) {
                            System.out.println("转发到死信队列失败:" + ex);
                        }

                        @Override
                        public void onSuccess(SendResult<String, Object> result) {
                            System.out.println("转发到死信队列成功:" + result.toString());
                        }
                    });
                });
            }

            /**
             * 不会到这方法，另外单独写的seek，无限循环
             */
            @Override
            public void handle(Exception thrownException, ConsumerRecords<?, ?> data, Consumer<?, ?> consumer, MessageListenerContainer container) {
                System.out.println(System.currentTimeMillis() + "--second here");
                Map<TopicPartition, Long> offsetsToReset = new HashMap<>(4);
                data.forEach(item -> {
                    System.out.println("batch error handler will seek min offset , data : " + item + " , exception : " + thrownException);
                    offsetsToReset.compute(new TopicPartition(item.topic(), item.partition()),
                            (k, v) -> v == null ? item.offset() : Math.min(v, item.offset()));
                });
                System.out.println("offsetsToReset" + offsetsToReset);
                offsetsToReset.forEach(consumer::seek);
            }
        };
    }


    @Bean
    @ConditionalOnMissingBean(BatchErrorHandler.class)
    public ErrorHandler errorHandler() {
        System.out.println("init errorHandler");
        return new ConsumerAwareErrorHandler() {
            @Override
            public void handle(Exception thrownException, ConsumerRecord<?, ?> data, Consumer<?, ?> consumer) {
                System.out.println("error handler will seek offset , data : " + data + " , exception : " + thrownException);
                consumer.seek(new TopicPartition(data.topic(), data.partition()), data.offset());
            }
        };
    }
}
